package top.lucky.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.List;
import java.util.Map;
import java.util.Properties;





public class KafkaTools {

    /**
     *
     * 私有静态方法，创建Kafka生产者
     *
     * @author IG
     * @Date 2017年4月14日 上午10:32:32
     * @version 1.0.0
     * @return KafkaProducer
     */
    private static KafkaProducer<String, String> createProducer() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "http://localhost:9195");// 声明kafka
        // properties.put("value.serializer",
        // "org.apache.kafka.common.serialization.ByteArraySerializer");
        // properties.put("key.serializer",
        // "org.apache.kafka.common.serialization.ByteArraySerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return new KafkaProducer<String, String>((properties));
    }

    /**
     *
     * 传入kafka约定的topicName,json格式字符串，发送给kafka集群
     *
     * @author IG
     * @Date 2017年4月14日 下午1:29:09
     * @version 1.0.0
     * @param topicName
     * @param jsonMessage
     */
    public static void sendMessage(String topicName, String jsonMessage) {
        KafkaProducer<String, String> producer = createProducer();
        producer.send(new ProducerRecord<String, String>(topicName, jsonMessage));
        producer.close();
    }

    /**
     *
     * 传入kafka约定的topicName,json格式字符串数组，发送给kafka集群<br>
     * 用于批量发送消息，性能较高。
     *
     * @author IG
     * @Date 2017年4月14日 下午2:00:12
     * @version 1.0.0
     * @param topicName
     * @param jsonMessages
     * @throws InterruptedException
     */
    public static void sendMessage(String topicName, String... jsonMessages) throws InterruptedException {
        KafkaProducer<String, String> producer = createProducer();
        for (String jsonMessage : jsonMessages) {
            producer.send(new ProducerRecord<String, String>(topicName, jsonMessage));
        }
        producer.close();
    }

    /**
     *
     * 传入kafka约定的topicName,Map集合，内部转为json发送给kafka集群 <br>
     * 用于批量发送消息，性能较高。
     *
     * @author IG
     * @Date 2017年4月14日 下午2:01:18
     * @version 1.0.0
     * @param topicName
     * @param mapMessageToJSONForArray
     */
    public static void sendMessage(String topicName, List<Map<Object, Object>> mapMessageToJSONForArray) {
        KafkaProducer<String, String> producer = createProducer();
        for (Map<Object, Object> mapMessageToJSON : mapMessageToJSONForArray) {
            String array = JacksonUtils.toJSONString(mapMessageToJSON);
            producer.send(new ProducerRecord<String, String>(topicName, array));
        }
        producer.close();
    }

    /**
     *
     * 传入kafka约定的topicName,Map，内部转为json发送给kafka集群
     *
     * @author IG
     * @Date 2017年4月14日 下午1:30:10
     * @version 1.0.0
     * @param topicName
     * @param mapMessageToJSON
     */
    public static void sendMessage(String topicName, Map<Object, Object> mapMessageToJSON) {
        KafkaProducer<String, String> producer = createProducer();
        String array = JacksonUtils.toJSONString(mapMessageToJSON);
        producer.send(new ProducerRecord<String, String>(topicName, array));
        producer.close();
    }


    public static void main(String[] args) throws InterruptedException {
 //       String[] s = new String[] { "{\"appId\":\"WDDCASH\",\"channelId\":\"CHANNEL_OPPO\",\"configId\":\"WDDCASH60V1\",\"fundRate\":\"0\",\"loanTime\":1619505915000,\"memberCode\":\"1855712\",\"outerTxnId\":\"200000371\",\"overdueFundRate\":\"0\",\"overdueYearRate\":\"0.24\",\"productId\":\"WDDCASH\",\"result\":1,\"txnId\":200000371,\"yearRate\":\"0.15\"}" };

   //     String[] s = new String[] {"{\"appId\":\"WDDCASH\",\"channelId\":\"CHANNEL_OPPO\",\"memberCode\":\"1855712\",\"outerTxnId\":\"200000372\",\"productId\":\"WDDCASH\",\"reason\":\"无可用资金源null\",\"result\":2,\"txnId\":200000372}"};

        String[] s = new String[] {"{ \"memberCode\":\"66600015144\", \"productId\":\"WDDCASH\", \"appId\":\"WDDCASH\", \"channelId\":\"CHANNEL_XIAODUI\", \"status\":2, \"riskLevel\": \"E\", \"auditRemark\" :  \"\", \"memo\" : \"额度激活\", \"limit\" : 3000, \"availLimit\" : 3000, \"auditTime\" : \"1652239946664\", \"blockCode\" : \"A\", \"startDate\":\"2022-05-11\",\"endDate\":\"2025-05-11\",\"kafkaUUID\":\"f9c685c0-3096-4fbd-b1e3-89540e055f08\", \"creditApplyId\":\"WDD155296725\", \"semsetimeScore\":\"0.9\" }\n"};
        // KafkaTools.sendMessage("logstest",
        // "{\"userName\":\"赵四\",\"pwd\":\"lisi\",\"age\":13}");
        /*
         * for (String a : s) { System.out.println(a); Thread.sleep(3000);
         * KafkaTools.sendMessage(topicName, jsonMessages); }
         */
        KafkaTools.sendMessage("first", s);
    }
}
